/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.snda.dw.lidejia.partitionfilegenerator;

import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.hive.serde2.binarysortable.OutputByteBuffer;

/**
 *
 * @author lidejia
 */
public class PartitionerMetaParser {

    private static interface MetaParser {

        public void parse(String data, boolean invert, OutputByteBuffer buffer);
    }

    private static class IntParser implements MetaParser {

        public void parse(String data, boolean invert, OutputByteBuffer buffer) {
            int v = Integer.parseInt(data);
            buffer.write((byte) ((v >> 24) ^ 0x80), invert);
            buffer.write((byte) (v >> 16), invert);
            buffer.write((byte) (v >> 8), invert);
            buffer.write((byte) v, invert);
        }
    }

    private static class LongParser implements MetaParser {

        public void parse(String data, boolean invert, OutputByteBuffer buffer) {
            long v = Long.parseLong(data);
            buffer.write((byte) ((v >> 56) ^ 0x80), invert);
            buffer.write((byte) (v >> 48), invert);
            buffer.write((byte) (v >> 40), invert);
            buffer.write((byte) (v >> 32), invert);
            buffer.write((byte) (v >> 24), invert);
            buffer.write((byte) (v >> 16), invert);
            buffer.write((byte) (v >> 8), invert);
            buffer.write((byte) v, invert);
        }
    }

    private static class StringParser implements MetaParser {

        public void parse(String dataStr, boolean invert, OutputByteBuffer buffer) {
            byte[] data = dataStr.getBytes();
            int length = dataStr.length();
            for (int i = 0; i < length; i++) {
                if (data[i] == 0 || data[i] == 1) {
                    buffer.write((byte) 1, invert);
                    buffer.write((byte) (data[i] + 1), invert);
                } else {
                    buffer.write(data[i], invert);
                }
            }
            buffer.write((byte) 0, invert);
        }
    }
    private Map<String, MetaParser> parsers = new HashMap<String, MetaParser>();

    private PartitionerMetaParser() {
        parsers.put("int", new IntParser());
        parsers.put("bigint", new LongParser());
        parsers.put("string", new StringParser());
    }

    public void parse(PartitionerMeta meta, String data, OutputByteBuffer buffer) {
        MetaParser parser = parsers.get(meta.getType());
        if (data == null || data.equals("")) {
            buffer.write((byte) 0, meta.isInvert());
            return;
        }
        // This field is not a null.
        buffer.write((byte) 1, meta.isInvert());
        parser.parse(data, meta.isInvert(), buffer);
    }
    private static final PartitionerMetaParser instance = new PartitionerMetaParser();

    public static PartitionerMetaParser getInstance() {
        return instance;
    }
}
